package com.niodata.dp.core.hdfs;

import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION;

import com.google.common.base.Strings;
import com.niodata.dp.core.ConfigurationKeys;
import com.niodata.dp.core.DpConfiguration;
import com.niodata.dp.core.resource.ResourceType.FileType;
import com.niodata.dp.core.usermgr.DpUsermgrException;
import com.niodata.dp.log.LogUtil;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Paths;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.io.FilenameUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.permission.AclEntry;
import org.apache.hadoop.fs.permission.AclEntryScope;
import org.apache.hadoop.fs.permission.AclEntryType;
import org.apache.hadoop.fs.permission.AclStatus;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.apache.log4j.Logger;


public class HdfsFileSystem {

  private static final HdfsFileSystem INSTANCE = new HdfsFileSystem();
  private static final Logger logger = Logger.getLogger(HdfsFileSystem.class);
  private final long maxCheckInterval = 10 * 60 * 1000;
  private UserGroupInformation ugi;
  private FileSystem fileSystem = null;
  private long lastLoginCheck = System.currentTimeMillis();

  private HdfsFileSystem() {
    System.setProperty("hadoop.home.dir",
          DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME));
    String userName =
          DpConfiguration.getString(ConfigurationKeys.HDFS_PROXY_USER);
    try {
      Configuration ugiConf = new Configuration();
      if (DpConfiguration.isHadoopKerberosEnabled()) {
        ugiConf.set(HADOOP_SECURITY_AUTHENTICATION, AuthenticationMethod.KERBEROS.name());
        UserGroupInformation.setConfiguration(ugiConf);
        System.setProperty("java.security.krb5.conf",
              DpConfiguration.getString(ConfigurationKeys.KERBEROS_CONF));
        String keytabPath =
              DpConfiguration.getString(ConfigurationKeys.HDFS_PROXY_USER_KEYTAB);
        ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(userName, keytabPath);
      } else {
        ugi = UserGroupInformation.createRemoteUser(userName);
      }
      fileSystem = doPrivileged(new PrivilegedExceptionAction<FileSystem>() {
        @Override
        public FileSystem run() throws Exception {
          Configuration conf = new Configuration();
          String coreSite =
                DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME)
                      + "/etc/hadoop/core-site.xml";
          String hdfsSite =
                DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME)
                      + "/etc/hadoop/hdfs-site.xml";
          conf.addResource(new Path(coreSite));
          conf.addResource(new Path(hdfsSite));
          return FileSystem.get(
                new URI(DpConfiguration.getString(ConfigurationKeys.HDFS_URI)), conf);
        }
      });
    } catch (DpUsermgrException | IOException | InterruptedException e) {
      LogUtil.logStackTrace(logger, e);
    }

  }

  public static HdfsFileSystem getInstance() {
    return INSTANCE;
  }

  public void setPermission(String path, final FsPermission permission) throws IOException {
    this.checkLogin();
    fileSystem.setPermission(new Path(path), permission);

  }


  public FileStatus getFileStatus(String path) throws IOException {
    this.checkLogin();
    return fileSystem.getFileStatus(new Path(path));
  }

  /**
   * 创建目录.
   *
   * @param path path
   * @param userName 用户名
   * @param groupName 组名
   * @throws DpUsermgrException 用户管理模块异常
   */
  public void mkdir(String path, String userName, String groupName)
        throws DpUsermgrException {
    try {
      doPrivileged(new PrivilegedExceptionAction<Boolean>() {
        @Override
        public Boolean run() throws IOException {
          String parent = Paths.get(path).getParent().toString();
          if (fileSystem.isFile(new Path(parent))) {
            throw new IOException("parent directory exists as a file.");
          }
          FsPermission fsPerm = new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE);
          if (fileSystem.mkdirs(new Path(path), fsPerm)) {
            if (!Strings.isNullOrEmpty(userName) && !Strings.isNullOrEmpty(groupName)) {
              fileSystem.setOwner(new Path(path), userName, groupName);
            }
            return true;
          }
          return false;
        }
      });
    } catch (IOException | InterruptedException e) {
      e.printStackTrace();
      throw new DpUsermgrException(500, "internal error");
    }
  }

  /**
   * 创建目录.
   *
   * @param path path
   * @param userName 用户名
   * @param groupName 组名
   * @throws DpUsermgrException 用户管理模块异常
   */
  public void mkdir(String path, String userName, String groupName, FsPermission permission)
        throws DpUsermgrException {
    try {
      doPrivileged(new PrivilegedExceptionAction<Boolean>() {
        @Override
        public Boolean run() throws IOException {
          String parent = Paths.get(path).getParent().toString();
          if (fileSystem.isFile(new Path(parent))) {
            throw new IOException("parent directory exists as a file.");
          }
          if (fileSystem.mkdirs(new Path(path), permission)) {
            if (!Strings.isNullOrEmpty(userName) && !Strings.isNullOrEmpty(groupName)) {
              fileSystem.setOwner(new Path(path), userName, groupName);
            }
            return true;
          }
          return false;
        }
      });
    } catch (IOException | InterruptedException e) {
      e.printStackTrace();
      throw new DpUsermgrException(500, "internal error");
    }
  }


  /**
   * push data from local to hdfs.
   */
  public void uploadData(String localPath, String remotePath)
        throws DpUsermgrException {
    try {
      doPrivileged(new PrivilegedExceptionAction<Boolean>() {
        @Override
        public Boolean run() throws IOException {
          fileSystem.copyFromLocalFile(new Path(localPath), new Path(remotePath));
          return true;
        }
      });
    } catch (IOException | InterruptedException e) {
      e.printStackTrace();
      throw new DpUsermgrException(500, "internal error");
    }
  }

  /**
   * push data from local to hdfs.
   */
  public void downloadData(String remotePath, String localPath)
        throws DpUsermgrException {
    try {
      doPrivileged(new PrivilegedExceptionAction<Boolean>() {
        @Override
        public Boolean run() throws IOException {
          fileSystem.copyToLocalFile(new Path(remotePath), new Path(localPath));
          return true;
        }
      });
    } catch (IOException | InterruptedException e) {
      e.printStackTrace();
      throw new DpUsermgrException(500, "internal error");
    }
  }

  /**
   * 删除目录.
   *
   * @param path 目录名
   * @throws DpUsermgrException 用户管理模块异常
   */
  public void deleteDir(String path) throws DpUsermgrException {
    try {
      doPrivileged(new PrivilegedExceptionAction<Void>() {
        @Override
        public Void run() throws Exception {
          String lastPattern = new Path(path).getName();
          FileStatus[] matchFiles = fileSystem
                .listStatus(new Path(path).getParent(), new PathFilter() {

                  @Override
                  public boolean accept(Path path) {
                    return FilenameUtils.wildcardMatch(path.getName(), lastPattern);
                  }
                });
          if (matchFiles != null && matchFiles.length > 0) {
            for (FileStatus file : matchFiles) {
              fileSystem.delete(file.getPath(), true);
            }
          }
          return null;
        }
      });
    } catch (IOException | InterruptedException e) {
      e.printStackTrace();
      throw new DpUsermgrException(500, "hdfs delete folder error");
    }
  }

  public boolean exist(String path) throws DpUsermgrException {
    try {
      this.checkLogin();
      return fileSystem.exists(new Path(path));
    } catch (IOException e) {
      throw new DpUsermgrException(500, "hdfs check folder exists error");
    }
  }

  /**
   * modifyAclEntries.
   */
  public void modifyAclEntries(String path, String name, AclEntryScope aclEntryScope,
        AclEntryType aclEntryType, String operationType) throws DpUsermgrException {
    try {
      this.checkLogin();
      AclEntry.Builder builder = new AclEntry.Builder();
      FsAction fsAction;
      if (operationType.equals(FileType.ReadWrite)) {
        fsAction = FsAction.ALL;
      } else {
        fsAction = FsAction.READ_EXECUTE;
      }
      AclEntry aclEntry = builder.setName(name).setType(aclEntryType).setScope(aclEntryScope)
            .setPermission(fsAction)
            .build();
      List<AclEntry> aclEntries = getEntries(path);
      aclEntries.add(aclEntry);
      fileSystem.modifyAclEntries(new Path(path), aclEntries);
    } catch (IOException e) {
      throw new DpUsermgrException(500, "hdfs set acl error", e);
    }
  }

  /**
   * modifyAclEntries.
   */
  public void modifyAclEntries(List<String> paths, String name, AclEntryScope aclEntryScope,
        AclEntryType aclEntryType, String operationType) throws DpUsermgrException {
    try {
      this.checkLogin();
      AclEntry.Builder builder = new AclEntry.Builder();
      FsAction fsAction;
      if (operationType.equals(FileType.ReadWrite)) {
        fsAction = FsAction.ALL;
      } else {
        fsAction = FsAction.READ_EXECUTE;
      }
      AclEntry aclEntry = builder.setName(name).setType(aclEntryType).setScope(aclEntryScope)
            .setPermission(fsAction)
            .build();
      for (String path : paths) {
        List<AclEntry> aclEntries = getEntries(path);
        aclEntries.add(aclEntry);
        fileSystem.modifyAclEntries(new Path(path), aclEntries);
      }
    } catch (IOException e) {
      throw new DpUsermgrException(500, "hdfs set acl error", e);
    }
  }

  public List<AclEntry> getEntries(String path) throws DpUsermgrException {
    try {
      this.checkLogin();
      AclStatus aclStatus = fileSystem.getAclStatus(new Path(path));
      return aclStatus.getEntries();
    } catch (IOException e) {
      throw new DpUsermgrException(500, "get acl status err", e);
    }
  }

  public AclStatus getAclStatus(String path) throws DpUsermgrException {
    try {
      this.checkLogin();
      AclStatus aclStatus = fileSystem.getAclStatus(new Path(path));
      return aclStatus;
    } catch (IOException e) {
      e.printStackTrace();
      throw new DpUsermgrException(500, "get acl status err", e);
    }
  }

  /**
   * removeAclEntries.
   */
  public void removeAclEntries(String path, String name, AclEntryScope aclEntryScope,
        AclEntryType aclEntryType, String operationType) throws DpUsermgrException {
    try {
      this.checkLogin();
      AclEntry.Builder builder = new AclEntry.Builder();
      FsAction fsAction;
      if (operationType.equals(FileType.ReadWrite)) {
        fsAction = FsAction.ALL;
      } else {
        fsAction = FsAction.READ_EXECUTE;
      }
      AclEntry aclEntry = builder.setName(name).setType(aclEntryType).setScope(aclEntryScope)
            .setPermission(fsAction)
            .build();
      List<AclEntry> aclEntries = new ArrayList<>();
      aclEntries.add(aclEntry);
      fileSystem.removeAclEntries(new Path(path), aclEntries);
    } catch (IOException e) {
      throw new DpUsermgrException(500, "hdfs set acl error", e);
    }
  }

  /**
   * removeAclEntries.
   */
  public void removeAclEntries(List<String> paths, String name, AclEntryScope aclEntryScope,
        AclEntryType aclEntryType, String operationType) throws DpUsermgrException {
    try {
      this.checkLogin();
      AclEntry.Builder builder = new AclEntry.Builder();
      FsAction fsAction;
      if (operationType.equals(FileType.ReadWrite)) {
        fsAction = FsAction.ALL;
      } else {
        fsAction = FsAction.READ_EXECUTE;
      }
      AclEntry aclEntry = builder.setName(name).setType(aclEntryType).setScope(aclEntryScope)
            .setPermission(fsAction)
            .build();
      for (String path : paths) {
        List<AclEntry> aclEntries = new ArrayList<>();
        aclEntries.add(aclEntry);
        fileSystem.removeAclEntries(new Path(path), aclEntries);
      }
    } catch (IOException e) {
      throw new DpUsermgrException(500, "hdfs set acl error", e);
    }
  }

  private <T> T doPrivileged(PrivilegedExceptionAction<T> action)
        throws DpUsermgrException, IOException, InterruptedException {
    if (ugi == null) {
      throw new DpUsermgrException(500, "hdfs login error");
    }
    try {
      ugi.checkTGTAndReloginFromKeytab();
    } catch (IOException e) {
      throw new DpUsermgrException(500, "hdfs login error");
    }
    return ugi.doAs(action);
  }


  private synchronized void checkLogin() throws IOException {
    if (!DpConfiguration.isHadoopKerberosEnabled()) {
      return;
    }
    long now = System.currentTimeMillis();
    long interval = now - this.lastLoginCheck;
    if (interval > this.maxCheckInterval) {
      this.ugi.checkTGTAndReloginFromKeytab();
      this.lastLoginCheck = now;
    }
  }

}
